package com.zx.mes.disruptor.hyl.multihignconsumer;

import com.lmax.disruptor.*;
import lombok.extern.java.Log;
import org.junit.Test;

import java.util.concurrent.*;
import java.util.stream.Stream;

/**
 * 多消费者 disruptor
 *
 * @author huayunlong
 * @date 2019-9-11
 */
@Log
public class Main {

    private static final int POOL_COUNT = Runtime.getRuntime().availableProcessors();

    public static void main(String[] args) throws InterruptedException {

        /// 1.线程池创建
        ThreadPoolExecutor executor = new ThreadPoolExecutor(POOL_COUNT / 3 + 1,
                POOL_COUNT + 1,
                120,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(200), r -> {
            Thread thread = new Thread(r);
            thread.setName("disruptor consumer thread");

            /// 对线程进行检测
            if (Thread.NORM_PRIORITY != thread.getPriority()) {
                thread.setPriority(Thread.NORM_PRIORITY);
            }
            return thread;
        }, (r, executor1) -> {
            if (!executor1.isShutdown()) {
                try {
                    BlockingQueue<Runnable> queue = executor1.getQueue();


                    int timeout = 500;
                    if (!queue.offer(r, timeout, TimeUnit.MILLISECONDS)) {
                        throw new RejectedExecutionException("Max wait time expired to queue task");
                    }

                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    throw new RejectedExecutionException("Interrupted", e);
                }
            } else {
                throw new RejectedExecutionException("Executor has been shut down");
            }
        });

        /// 2.ringBuffer
        RingBuffer<Order> ringBuffer = RingBuffer.createMultiProducer(Order::new,
                1024 * 1024,
                new SleepingWaitStrategy());

        /// 3.能过ringBuffer创建一个屏障
        SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();

        /// 4.创建消费者
        Consumer[] consumers = new Consumer[5];
        for (int i = 0; i < consumers.length; i++) {
            consumers[i] = new Consumer("H" + i);
        }

        /// 5.构建消费者工作池
        WorkerPool<Order> workerPool = new WorkerPool<>(ringBuffer,
                sequenceBarrier,
                new MyExceptionHandler(),
                consumers);

        /// 6.设置多个消费者sequence序号,用于单独统计消费进度,并设置至ringBuffer
        ringBuffer.addGatingSequences(workerPool.getWorkerSequences());

        /// 7.启动workPool
        workerPool.start(executor);

        CountDownLatch latch = new CountDownLatch(1);

        /// 8.投递任务
        for (int i = 0; i <100 ; i++) {
            final Producer producer = new Producer(ringBuffer);

            new Thread(() -> {
                try {
                    latch.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                for (int j = 0; j <100 ; j++) {
                    producer.send();
                }

            }).start();
        }
        Thread.sleep(1000);
        /// 数据创建完毕
        System.err.println("----------线程创建完毕，开始生产数据----------");
        latch.countDown();
        Thread.sleep(3000);
///        System.err.println("任务总数:" + consumers[1].getCount());
        Stream.of(consumers).forEach(consumer -> {
            System.out.println(consumer.getCount());
        });
        executor.shutdown();
//        workerPool.halt();
    }


    @Test
    public void test() {
        Consumer[] consumers = new Consumer[5];
//        Stream.of(consumers).forEach((consumer -> {
//            consumer = new Consumer("H");
//        }));
        final Stream<Object> objectStream = Stream.of(consumers).map((consumer) -> {
            System.out.println(" 123456 ");
            return null;
        });
        System.out.println(objectStream);
    }
}

class MyExceptionHandler implements ExceptionHandler<Order> {

    @Override
    public void handleEventException(Throwable ex, long sequence, Order event) {
        System.out.println("event:" + event.toString() + " : " + "sequence:" + sequence);
    }

    @Override
    public void handleOnStartException(Throwable ex) {
        System.out.println("消费者处理开始异常!");
    }

    @Override
    public void handleOnShutdownException(Throwable ex) {
        System.out.println("消费者关闭异常!");
    }
}
